|
|
@@ -13,10 +13,7 @@ protocol NightscoutManager: GlucoseSource {
|
|
|
func deleteCarbs(withID id: String) async
|
|
|
func deleteInsulin(withID id: String) async
|
|
|
func deleteGlucose(withID id: String, withDate date: Date) async
|
|
|
- func uploadDeviceStatus() async throws
|
|
|
- func uploadGlucose() async
|
|
|
func uploadCarbs() async
|
|
|
- func uploadPumpHistory() async
|
|
|
func uploadOverrides() async
|
|
|
func uploadTempTargets() async
|
|
|
func uploadProfiles() async throws
|
|
|
@@ -44,56 +41,33 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
private let processQueue = DispatchQueue(label: "BaseNetworkManager.processQueue")
|
|
|
private var ping: TimeInterval?
|
|
|
|
|
|
- // Queue where upload pipelines run.
|
|
|
- let uploadPipelineQueue = DispatchQueue(label: "NightscoutManager.uploadPipelines", qos: .utility)
|
|
|
-
|
|
|
- /// Throttle window (seconds) per upload pipeline. Any requests inside this window
|
|
|
- /// coalesce into a single upload run for that pipeline.
|
|
|
- let uploadPipelineInterval: [NightscoutUploadPipeline: TimeInterval] = [
|
|
|
- .carbs: 2, .pumpHistory: 2, .overrides: 2, .tempTargets: 2,
|
|
|
- .glucose: 2, .deviceStatus: 2
|
|
|
- ]
|
|
|
-
|
|
|
- /// Subjects used to request an upload pipeline. The pipeline applies a throttle so
|
|
|
- /// close calls don’t double-upload.
|
|
|
- var uploadPipelineSubjects: [NightscoutUploadPipeline: PassthroughSubject<Void, Never>] = {
|
|
|
- var d: [NightscoutUploadPipeline: PassthroughSubject<Void, Never>] = [:]
|
|
|
- NightscoutUploadPipeline.allCases.forEach { d[$0] = PassthroughSubject<Void, Never>() }
|
|
|
- return d
|
|
|
- }()
|
|
|
+ /// Coalesces and serializes upload runs so no two runs of the same pipeline overlap.
|
|
|
+ /// Runs execute `performUpload(for:)`, provided at init. The public `upload*()`
|
|
|
+ /// methods and `requestUpload(_:)` go through the serializer; the `performUpload*()`
|
|
|
+ /// bodies must not call the awaitable `upload*()` entry points. The serializer
|
|
|
+ /// asserts on such a call in debug builds and downgrades it to a fire-and-forget
|
|
|
+ /// request in release.
|
|
|
+ private var uploadSerializer: NightscoutUploadSerializer!
|
|
|
|
|
|
/// Request an upload for a pipeline (enqueue work). Safe to call from anywhere.
|
|
|
+ /// Bursts of requests coalesce: at most one run follows the one currently in flight.
|
|
|
func requestUpload(_ uploadPipeline: NightscoutUploadPipeline) {
|
|
|
- uploadPipelineSubjects[uploadPipeline]?.send(())
|
|
|
- }
|
|
|
-
|
|
|
- /// Build the Combine pipelines for all upload pipelines: subject → throttle → upload.
|
|
|
- /// Must be called once during init().
|
|
|
- func setupLanePipelines() {
|
|
|
- for pipeline in NightscoutUploadPipeline.allCases {
|
|
|
- guard let subject = uploadPipelineSubjects[pipeline], let window = uploadPipelineInterval[pipeline] else { continue }
|
|
|
- subject
|
|
|
- .receive(on: uploadPipelineQueue)
|
|
|
- .throttle(for: .seconds(window), scheduler: uploadPipelineQueue, latest: false)
|
|
|
- .sink { [weak self] in
|
|
|
- guard let self else { return }
|
|
|
- Task(priority: .utility) { await self.runUploadPipeline(pipeline) }
|
|
|
- }
|
|
|
- .store(in: &subscriptions)
|
|
|
+ Task(priority: .utility) { [weak self] in
|
|
|
+ await self?.uploadSerializer.request(uploadPipeline)
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- /// Runs the actual upload for a single upload pipeline.
|
|
|
- /// Called by the throttled pipeline, not directly by callers.
|
|
|
- func runUploadPipeline(_ uploadPipeline: NightscoutUploadPipeline) async {
|
|
|
+ /// Dispatches to a pipeline's upload routine. Runs inside the serializer; must only
|
|
|
+ /// call `performUpload*()` functions, never the public `upload*()` entry points.
|
|
|
+ private func performUpload(for uploadPipeline: NightscoutUploadPipeline) async {
|
|
|
switch uploadPipeline {
|
|
|
- case .carbs: await uploadCarbs()
|
|
|
- case .pumpHistory: await uploadPumpHistory()
|
|
|
- case .overrides: await uploadOverrides()
|
|
|
- case .tempTargets: await uploadTempTargets()
|
|
|
- case .glucose: await uploadGlucose()
|
|
|
+ case .carbs: await performUploadCarbs()
|
|
|
+ case .pumpHistory: await performUploadPumpHistory()
|
|
|
+ case .overrides: await performUploadOverrides()
|
|
|
+ case .tempTargets: await performUploadTempTargets()
|
|
|
+ case .glucose: await performUploadGlucose()
|
|
|
case .deviceStatus:
|
|
|
- do { try await uploadDeviceStatus() }
|
|
|
+ do { try await performUploadDeviceStatus() }
|
|
|
catch { debug(.nightscout, "deviceStatus upload failed: \(error)") }
|
|
|
}
|
|
|
}
|
|
|
@@ -134,14 +108,11 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
|
|
|
// MARK: - Upload triggers
|
|
|
|
|
|
- //
|
|
|
// Each upload pipeline is driven by an NSFetchedResultsController whose predicate is the
|
|
|
// "not yet uploaded to Nightscout" set for that entity. The controller fires whenever
|
|
|
// un-uploaded items appear (or drop out after a successful upload), which we map to a
|
|
|
- // `requestUpload(pipeline)` call (throttled per pipeline). Bound to the viewContext, they
|
|
|
- // also pick up batch-inserted glucose via the persistent history merge in CoreDataStack —
|
|
|
- // replacing the previous changedObjects publisher plus the glucoseStorage.updatePublisher
|
|
|
- // fallback.
|
|
|
+ // `requestUpload(pipeline)` call (coalesced and serialized per pipeline). Bound to the viewContext,
|
|
|
+ // they also pick up batch-inserted glucose via the persistent history merge in CoreDataStack.
|
|
|
|
|
|
let determinationUploadControllerDelegate = FetchedResultsControllerDelegate()
|
|
|
lazy var determinationUploadController: NSFetchedResultsController<OrefDetermination> = {
|
|
|
@@ -293,16 +264,20 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
|
|
|
init(resolver: Resolver) {
|
|
|
injectServices(resolver)
|
|
|
+
|
|
|
+ uploadSerializer = NightscoutUploadSerializer { [weak self] pipeline in
|
|
|
+ await self?.performUpload(for: pipeline)
|
|
|
+ }
|
|
|
+
|
|
|
subscribe()
|
|
|
|
|
|
setupNotification()
|
|
|
|
|
|
- setupLanePipelines()
|
|
|
wireSubscribers()
|
|
|
|
|
|
/// Ensure that Nightscout Manager holds the `lastEnactedDetermination`, if one exists, on initialization.
|
|
|
/// We have to set this here in `init()`, so there's a `lastEnactedDetermination` available after an app restart
|
|
|
- /// for `uploadDeviceStatus()`, as within that fuction `lastEnactedDetermination` is reassigned at the very end of the function.
|
|
|
+ /// for `performUploadDeviceStatus()`, as within that function `lastEnactedDetermination` is reassigned at the very end of the function.
|
|
|
/// This way, we ensure the latest enacted determination is always part of `devicestatus` and avoid having instances
|
|
|
/// where the first uploaded non-enacted determination (i.e., "suggested"), lacks the "enacted" data.
|
|
|
Task {
|
|
|
@@ -327,6 +302,9 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /// Direct upload triggers for App Intents / remote control, which await the upload
|
|
|
+ /// before ending their background task. Uploads are serialized per pipeline, so these
|
|
|
+ /// may fire alongside the Core Data triggers for the same change without double-uploading.
|
|
|
func setupNotification() {
|
|
|
Foundation.NotificationCenter.default.publisher(for: .willUpdateOverrideConfiguration)
|
|
|
.sink { [weak self] _ in
|
|
|
@@ -554,8 +532,7 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
/// - Schedule a task to upload pod age data separately.
|
|
|
///
|
|
|
/// - Note: Ensure `nightscoutAPI` is initialized and `isUploadEnabled` is set to `true` before invoking this function.
|
|
|
- /// - Returns: Nothing.
|
|
|
- func uploadDeviceStatus() async throws {
|
|
|
+ private func performUploadDeviceStatus() async throws {
|
|
|
guard let nightscout = nightscoutAPI, isUploadEnabled else {
|
|
|
debug(.nightscout, "NS API not available or upload disabled. Aborting NS Status upload.")
|
|
|
return
|
|
|
@@ -914,7 +891,7 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- func uploadGlucose() async {
|
|
|
+ private func performUploadGlucose() async {
|
|
|
do {
|
|
|
try await uploadGlucose(glucoseStorage.getGlucoseNotYetUploadedToNightscout())
|
|
|
try await uploadNonCoreDataTreatments(glucoseStorage.getCGMStateNotYetUploadedToNightscout())
|
|
|
@@ -926,7 +903,7 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- func uploadPumpHistory() async {
|
|
|
+ private func performUploadPumpHistory() async {
|
|
|
do {
|
|
|
try await uploadPumpHistory(pumpHistoryStorage.getPumpHistoryNotYetUploadedToNightscout())
|
|
|
} catch {
|
|
|
@@ -938,6 +915,10 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
}
|
|
|
|
|
|
func uploadCarbs() async {
|
|
|
+ await uploadSerializer.run(.carbs)
|
|
|
+ }
|
|
|
+
|
|
|
+ private func performUploadCarbs() async {
|
|
|
do {
|
|
|
try await uploadCarbs(carbsStorage.getCarbsNotYetUploadedToNightscout())
|
|
|
try await uploadCarbs(carbsStorage.getFPUsNotYetUploadedToNightscout())
|
|
|
@@ -950,6 +931,10 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
}
|
|
|
|
|
|
func uploadOverrides() async {
|
|
|
+ await uploadSerializer.run(.overrides)
|
|
|
+ }
|
|
|
+
|
|
|
+ private func performUploadOverrides() async {
|
|
|
do {
|
|
|
try await uploadOverrides(overridesStorage.getOverridesNotYetUploadedToNightscout())
|
|
|
try await uploadOverrideRuns(overridesStorage.getOverrideRunsNotYetUploadedToNightscout())
|
|
|
@@ -962,6 +947,10 @@ final class BaseNightscoutManager: NightscoutManager, Injectable {
|
|
|
}
|
|
|
|
|
|
func uploadTempTargets() async {
|
|
|
+ await uploadSerializer.run(.tempTargets)
|
|
|
+ }
|
|
|
+
|
|
|
+ private func performUploadTempTargets() async {
|
|
|
do {
|
|
|
try await uploadTempTargets(await tempTargetsStorage.getTempTargetsNotYetUploadedToNightscout())
|
|
|
try await uploadTempTargetRuns(await tempTargetsStorage.getTempTargetRunsNotYetUploadedToNightscout())
|
Deniz Cengiz